package InjectData;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.log4j.Logger;

import com.vccorp.ads.model.AdsLogData;


public class InjectDataToHDFS {
	private Configuration conf;
	private Logger logger=Logger.getLogger("Main");
	/**
	 * add Inject file to hdfs
	 * 
	 * @param source
	 * @param dest
	 * @throws IOException
	 */
	public InjectDataToHDFS() {
		// TODO Auto-generated constructor stub
		conf=new Configuration();
	}
	public void addFile(String source) throws IOException {
		String[] split=source.split("/");
		String dest="AdsLog2013"+"/"+split[split.length-2]+"/"+split[split.length-1].substring(3,15)+".seq";
		logger.info(dest);
		FileSystem fs = FileSystem.get(conf);
		Path path = new Path(dest);
		if(fs.exists(path))
		{
			logger.info("File Existed!");
		}
		NullWritable key = NullWritable.get();
		AdsLogData value = new AdsLogData();
		SequenceFile.Writer writer = null;
		BufferedReader br = new BufferedReader(new FileReader(new File(source)));
		String line = null;
		writer = SequenceFile.createWriter(fs, conf, path, key.getClass(),
				value.getClass());
		while ((line = br.readLine()) != null) {
			value = new AdsLogData(line);
			writer.append(key,value);
		}
		writer.close();
	}

	/**
	 * list file from Folder
	 * 
	 * @param folder
	 * @return
	 */
	public List<String> listFilesForFolder(final File folder) {
		List<String> listFileName = new ArrayList<String>();
		for (final File fileEntry : folder.listFiles()) {
			if (fileEntry.isDirectory()) {
				listFilesForFolder(fileEntry);
			} else {
				listFileName.add(fileEntry.getPath());
			}
		}
		return listFileName;
	}
	public static void main(String[] args) {
		InjectDataToHDFS data=new InjectDataToHDFS();
		final File folder=new File("D:/Data");
		List<String> listFile=data.listFilesForFolder(folder);
		for (String fileName : listFile) {
			System.out.println(fileName);
		}
	}
}
